package com.nami.logic.rocketmq.consumer;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import cn.hutool.json.JSONUtil;
import com.nami.basic.constants.DownTopicConst;
import com.nami.basic.constants.UpTopicConst;
import com.nami.basic.request.upmsg.ChannelCloseMqWrap;
import com.nami.logic.service.DownMsgSender;
import com.nami.logic.service.PacketFactory;
import com.nami.logic.service.UserService;
import com.nami.store.api.bo.UserDevice;
import com.nami.store.api.bo.UserWrap;
import com.nami.store.api.request.cache.CacheDelReq;
import com.nami.store.api.service.RemoteL2CacheService;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.dromara.common.core.utils.IdentityUtil;
import org.springframework.stereotype.Component;

import java.util.List;

@Slf4j
@Component
@AllArgsConstructor
@RocketMQMessageListener(topic = UpTopicConst.CONNECT_TIMEOUT_TOPIC,
    consumerGroup = UpTopicConst.CONNECT_TIMEOUT_TOPIC,
    consumeMode = ConsumeMode.CONCURRENTLY)
public class ConnectTimeoutConsumer implements RocketMQListener<String> {

    private final DownMsgSender mqSender;
    private final PacketFactory packetFactory;

    private final UserService userService;

    @DubboReference
    private final RemoteL2CacheService cacheService;

    @Override
    public void onMessage(String message) {
        try {
            log.info("[connect-consumer] begin. message:{}", message);
            TimeInterval timer = DateUtil.timer();
            if (StringUtils.isBlank(message)) {
                log.warn("[connect-consumer] message is empty");
                return;
            }
            ChannelCloseMqWrap body = JSONUtil.toBean(message, ChannelCloseMqWrap.class);

            if (body == null) {
                log.warn("[connect-consumer] message is empty");
                return;
            }
            if (StringUtils.isBlank(body.getAppKey())
                || StringUtils.isBlank(body.getUserId())
                || body.getDeviceType() == null
            ) {
                log.warn("[connect-consumer] param is empty.");
                return;
            }

            UserWrap userFromCache = userService.getOnrUserById(body.getAppKey(), body.getUserId());
            if (userFromCache == null) {
                log.warn("[connect-consumer] 缓存不在此节点.");
                return;
            }
            List<UserDevice> deviceInfos = userFromCache.getDeviceInfos();
            if (CollUtil.isEmpty(deviceInfos)) {
                log.warn("[connect-consumer] 当前设备为空.");
                return;
            }
            UserDevice deviceInfo = deviceInfos.stream().filter(item -> item != null && body.getDeviceType().equals(item.getDeviceType())).findFirst().orElse(null);
            if (deviceInfo == null) {
                log.warn("[connect-consumer] 缓存不在此节点.");
                return;
            }

            // TODO 发mq广播消息，通知所有logic服务 删除一级缓存得user当前在线设备数据
            // TODO 如果是改为了user服务，那么则是发送mq 广播到所有user服务
            CacheDelReq delReq = new CacheDelReq();
            delReq.setAppKey(body.getAppKey());
            delReq.setUserId(body.getUserId());
            delReq.setDeviceType(body.getDeviceType());
            userService.removeOfflineUser(delReq);

            mqSender.syncSendByMq(DownTopicConst.CLOSE_CHANNEL_TOPIC, deviceInfo.getMqTags(), IdentityUtil.buildIdentityNo(body.getUserId(), body.getDeviceType()));
        } catch (Exception var3) {
            var3.printStackTrace();
            log.error("timeout error:", var3);
        }

    }


}
